package com.example.ffepeg.ffeAdder;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class QpsLimitedThreadPool {
    private static final Logger log = LoggerFactory.getLogger(QpsLimitedThreadPool.class);
    private final Semaphore semaphore;
    private final int maxQps;
    private final AtomicInteger availableTokens;
    private final ScheduledExecutorService scheduler;
    private static final ThreadPoolExecutor executor;

    public ThreadPoolExecutor createPool(){  //创建多线程
        return executor;
    }

    static {
        executor =new ThreadPoolExecutor(
                10,         // 核心线程数
                20,                   // 最大线程数
                1,                    // 线程空闲时间
                TimeUnit.MINUTES,     // 时间单位
                new LinkedBlockingQueue<>(100),  // 工作队列
                Executors.defaultThreadFactory(), // 默认线程工厂
                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
            );
    }

    public QpsLimitedThreadPool(int maxQps) {
        this.maxQps = maxQps;
        this.semaphore = new Semaphore(maxQps, true); // 公平锁
        this.availableTokens = new AtomicInteger(maxQps);
        this.scheduler = Executors.newSingleThreadScheduledExecutor();
        this.scheduler.scheduleAtFixedRate(this::addToken, 0, 1, TimeUnit.SECONDS);
    }
    private void addToken() {
        while (availableTokens.get() < maxQps && !availableTokens.compareAndSet(maxQps, maxQps)) {
            if (availableTokens.incrementAndGet() <= maxQps) {
                semaphore.release(1);
            }
        }
    }
    public void shutdown() {   //关闭QPS与多线程的
        // 关闭定时器
        scheduler.shutdown();
        // 平滑关闭线程池
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) { // 等待60秒
                // 如果超过60秒还没有结束，则强制关闭
                List<Runnable> notExecutedTasks = executor.shutdownNow();
                log.info("立即关闭，不执行任务: " + notExecutedTasks.size());
            }
        } catch (InterruptedException e) {
            // 如果等待过程中发生中断，强制关闭线程池
            executor.shutdownNow();
            Thread.currentThread().interrupt(); // 保留中断状态
        } catch (Exception e){
            throw new RuntimeException(e.getMessage());
        }
    }
    //执行QPS的方法调用
    public void execute(Runnable task) throws InterruptedException {
        semaphore.acquire();
        try {
            executor.execute(() -> {
                try {
                    task.run();
                } finally {
                    semaphore.release();
                }
            });
        } catch (RejectedExecutionException e) {
            semaphore.release();
            throw e;
        }
    }

    private void shutdownExecutor() {  //关闭多线程的
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                log.error("部分任务未完成，强制关闭...");
                executor.shutdownNow();

                if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                    log.error("线程池关闭失败");
                }
            }
        } catch (InterruptedException ie) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        } catch (Exception e){
            throw new RuntimeException(e);
        }
    }

}
